package test.disruptor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.Executors;

public class FromIpush {
    private static final Logger logger = LoggerFactory.getLogger(FromIpush.class);

    @Test
    public void test() throws InterruptedException {

        RingBuffer<ValueEvent> buffer = buildConsumer();

        logger.info("开始生产...");
        for (int i = 0; i < 1000; i++) {
            produce(buffer, i);
        }

        Thread.sleep(Long.MAX_VALUE);
    }

    /**
     * 构造消费者
     * @return
     */
    public  RingBuffer<ValueEvent> buildConsumer()  {
        //定义一个ringBuffer,也就是相当于一个队列
        RingBuffer<ValueEvent> buffer = RingBuffer.create(ProducerType.MULTI, ValueEvent.EVENT_FACTORY, 1024, new BlockingWaitStrategy());
        // 定义消费者,只要有生产出来的东西，该事件就会被触发
        // 参数 event 为被生产出来的东西　
        // 几个workerHandler 表示有几个消费者
        WorkHandler<ValueEvent> workHandler = event -> {
            String value = event.getValue();
            logger.info("[Thread]:{},value:{}",Thread.currentThread().getName(),value);
        };

        //定义一个消费者池，每一个handler都是一个消费者，也就是一个线程，会不停地就队列中请求位置，如果这们位置中被生产者放入了东西，而这个东西则是上面定义
        //RingBuffer中的 factory 创建出来的一个event,消费者会取出这个event,调用handler中的onEvent方法，如果这个位置还没有被生产者放入东西，则阻塞，等待生产者生产后唤醒.
        //而生产者在生产时要先申请slot，而这个slot位置不能高于最后一个消费者的位置，否则会覆盖没有消费的slot，如果大于消费者的最后一个slot，则进行自旋等待.
        WorkerPool<ValueEvent> workerPool = new WorkerPool<>(buffer,
                buffer.newBarrier(), new IgnoreExceptionHandler(), workHandler, workHandler, workHandler, workHandler);
        //每个消费者，也就是 workProcessor都有一个sequence，表示上一个消费的位置,这个在初始化时都是-1
        Sequence[] sequences = workerPool.getWorkerSequences();
        //将其保存在ringBuffer中的 sequencer 中，在为生产申请slot时要用到,也就是在为生产者申请slot时不能大于此数组中的最小值,否则产生覆盖
        buffer.addGatingSequences(sequences);
        workerPool.start(Executors.newFixedThreadPool(10));			//用executor 来启动 workProcessor 线程
        logger.info("disruptor started！");
        return buffer;
    }

    private static void produce(RingBuffer<ValueEvent> buffer, int i) {
        /*
         * 对于生产者生产都是以 下面方式
         *
         * 		long next = ringBuffer.next();
         *
         * 			try{
         * 				Event event = ringBuffer.get(next);
         * 				event.doxxx		//相当于生产
         * 			}finally{
         * 				ringBuffer.publis(next);		//将slot 发布，必须
         * 			}
         *
         *
         * */
            long next = buffer.next();
            try {
                ValueEvent event = buffer.get(next);
                event.setValue(""+i);
            } finally {
                logger.info("生产:{}",i);
                buffer.publish(next);
            }

    }

}